package net.bwie.realtime.jtp.dwd.log.function;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/05/18 22:20:13
 * @version: 1.0
 */
public class LogSplitProcessFunction extends ProcessFunction<String,String>{

    private final OutputTag<String> errorTag;
    private final OutputTag<String> startTag;
    private final OutputTag<String> displayTag;
    private final OutputTag<String> actionTag;

    public LogSplitProcessFunction(OutputTag<String> errorTag, OutputTag<String> startTag, OutputTag<String> displayTag, OutputTag<String> actionTag) {
        this.errorTag = errorTag;
        this.startTag = startTag;
        this.displayTag = displayTag;
        this.actionTag = actionTag;
    }

            @Override
            public void processElement(String value, Context ctx, Collector<String> out) throws Exception {

                JSONObject jsonObject = JSON.parseObject(value);

                if (null != jsonObject.get("err")) {
                    ctx.output(errorTag, value);
                    jsonObject.remove("err");
                }

                if (null != jsonObject.get("start")) {
                    ctx.output(startTag, jsonObject.toJSONString());
                }

                else {
//                    曝光 动作
                    JSONArray displaysArray = jsonObject.getJSONArray("displays");
                    JSONArray actionsArray = jsonObject.getJSONArray("actions");

                    jsonObject.remove("displays");
                    jsonObject.remove("actions");

                    out.collect(jsonObject.toJSONString());


                    if (null != displaysArray && !displaysArray.isEmpty()) {
                        for (int i = 0; i < displaysArray.size(); i++) {
                            JSONObject displaysArrayJSONObject = displaysArray.getJSONObject(i);
                            jsonObject.put("display", displaysArrayJSONObject);
                            ctx.output(displayTag, jsonObject.toJSONString());
                        }
                    }
                    jsonObject.remove("display");

                    if (null != actionsArray && !actionsArray.isEmpty()) {
                        for (int i = 0; i < actionsArray.size(); i++) {
                            JSONObject actionsArrayJSONObject = actionsArray.getJSONObject(i);
                            jsonObject.put("action", actionsArrayJSONObject);
                            ctx.output(actionTag, jsonObject.toJSONString());
                        }
                    }
                }
            }
        };


